/**
 * Copyright (c) 2013, Andrea Funto'. All rights reserved.
 * 
 * This file is part of the Dihedron Concurrent Utilities library ("Concurrent").
 *
 * "Commons" is free software: you can redistribute it and/or modify it under 
 * the terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version.
 *
 * "Concurrent" is distributed in the hope that it will be useful, but WITHOUT ANY 
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 
 * A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more 
 * details.
 *
 * You should have received a copy of the GNU Lesser General Public License 
 * along with "Concurrent". If not, see <http://www.gnu.org/licenses/>.
 */

package org.dihedron.concurrent;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The class used to start new tasks asynchronously and then wait for their 
 * completion; it makes use of TaskExecutors to start tasks, so an asynchronous
 * execution can be spawned by means of a thread, an ExecutorService or an
 * asynchronous EJB method. 
 * 
 * @author Andrea Funto'
 *
 * @param <T>
 *   the type of objects returned by new asynchronous tasks.
 */
public class TaskService<T> {
	
	/**
	 * The logger.
	 */
	private final static Logger logger = LoggerFactory.getLogger(TaskService.class);
	
	/**
	 * The actual executor, whether an ExecutorService or an asynchronous EJB.
	 */
	private TaskExecutor<T> executor;
	
	/**
	 * The queue used to synchronise task execution.
	 */
	private BlockingQueue<Integer> queue;
	
	/**
	 * Constructor.
	 *
	 * @param executor
	 *   the actual service that will take care of spawning the asynchronous 
	 *   thread of execution; this object can be an adapter for an ExecutorService
	 *   or for an EJB 3.1 asynchornous method.
	 */
	public TaskService(TaskExecutor<T> executor) {
		this.executor = executor;
		this.queue = new LinkedBlockingQueue<Integer>();
	}
	
	/**
	 * Starts the given set of tasks asynchronously, returning their futures.
	 * 
	 * @param tasks
	 *   the set of tasks to execute.
	 * @return
	 *   the corresponding set of futures.
	 */
	public List<Future<T>> execute(Task<T>... tasks) {
		List<Future<T>> futures = new ArrayList<Future<T>>();
		int i = 0;
		for(Task<T> task : tasks) {
			task.setId(i++);
			task.setQueue(queue);
			futures.add(executor.submit(task));
		}
		return futures;
	}
	
	/**
	 * Starts the given set of tasks asynchronously, returning their futures.
	 * 
	 * @param tasks
	 *   the set of tasks to execute.
	 * @return
	 *   the corresponding set of futures.
	 */
	public List<Future<T>> execute(List<Task<T>> tasks) {
		List<Future<T>> futures = new ArrayList<Future<T>>();
		int i = 0;
		for(Task<T> task : tasks) {
			task.setId(i++);
			task.setQueue(queue);
			futures.add(executor.submit(task));
		}
		
		return futures;
	}
	
	/**
	 * Waits for all tasks to complete before returning; if observers are provided,
	 * they are called each time a task completes.
	 * 
	 * @param futures
	 *   the list of futures to wait for.
	 * @param observers
	 *   an optional set of observers.
	 * @return
	 *   the list of results, once all tasks are done.
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TaskTimedOutException 
	 */
	public List<T> waitForAll(List<Future<T>> futures, TaskObserver<T>... observers) throws InterruptedException, ExecutionException, TaskTimedOutException {
		return waitForAll(futures, Long.MAX_VALUE, TimeUnit.DAYS, observers);
	}
	
	/**
	 * Waits for all tasks to complete before returning; if observers are provided,
	 * they are called each time a task completes.
	 * 
	 * @param futures
	 *   the list of futures to wait for.
	 * @param timeout
	 *   the time to wait, in <code>unit</code> time units.
	 * @param unit
	 *   the time unit of the previous parameter.
	 * @param observers
	 *   an optional set of observers.
	 * @return
	 *   the list of results, once all tasks are done.
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TaskTimedOutException 
	 */
	public List<T> waitForAll(List<Future<T>> futures, long timeout, TimeUnit unit, TaskObserver<T>... observers) throws InterruptedException, ExecutionException, TaskTimedOutException {
		Map<Integer, T> results = new HashMap<Integer, T>();
		int count = futures.size();
		//timeout = TimeUnit.MILLISECONDS.convert(timeout, unit); 
		while(count-- > 0) {
			long start = System.currentTimeMillis();
			Integer id = queue.poll(timeout, unit);
			timeout = timeout - unit.convert((System.currentTimeMillis() - start), TimeUnit.MILLISECONDS);
			if(id == null || timeout < 0) {
				logger.trace("no tasks completed within timeout, exiting");
				throw new TaskTimedOutException("timeout expired before all tasks completed");
			}
			logger.debug("task '{}' complete (count: {}, queue: {})", id, count, queue.size());
			T result = futures.get(id).get();
			results.put(id, result);
			for(TaskObserver<T> observer : observers) {
				observer.onTaskComplete(result);
			}
		}
		
		logger.debug("all tasks complete");
		List<T> values = new ArrayList<T>();
		for(int i = 0; i < results.size(); ++i) {
			values.add(results.get(i));
		}
		return values;
	}	

	/**
	 * Waits until the first task completes, then calls the (optional) observers 
	 * to notify the completion and returns the result.
	 * 
	 * @param futures
	 *   the list of futures to wait for.
	 * @param observers
	 *   an optional set of observers.
	 * @return
	 *   the result of the first task to complete.
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public T waitForAny(List<Future<T>> futures, TaskObserver<T>... observers) throws InterruptedException, ExecutionException {
		int count = futures.size();
		while(count-- > 0) {
			int id = queue.take();
			logger.debug("task '{}' complete (count: {}, queue: {})", id, count, queue.size());
			T result = futures.get(id).get();
			for(TaskObserver<T> observer : observers) {
				observer.onTaskComplete(result);
			}
			return result;
		}
		return null;
	}

	/**
	 * Waits until the first task completes, then calls the (optional) observers 
	 * to notify the completion and returns the result.
	 * 
	 * @param futures
	 *   the list of futures to wait for.
	 * @param timeout
	 *   the time to wait, in <code>unit</code> time units.
	 * @param unit
	 *   the time unit of the previous parameter.   
	 * @param observers
	 *   an optional set of observers.
	 * @return
	 *   the result of the first task to complete.
	 * @throws InterruptedException
	 * @throws ExecutionException
	 * @throws TaskTimedOutException 
	 */
	public T waitForAny(List<Future<T>> futures, long timeout, TimeUnit unit, TaskObserver<T>... observers) throws InterruptedException, ExecutionException, TaskTimedOutException {
		int count = futures.size();
		//timeout = TimeUnit.MILLISECONDS.convert(timeout,  unit);
		while(count-- > 0 && timeout > 0) {
			long start = System.currentTimeMillis();
			Integer id = queue.poll(timeout, unit);
			//timeout = timeout - (System.currentTimeMillis() - start);
			timeout = timeout - unit.convert((System.currentTimeMillis() - start), TimeUnit.MILLISECONDS);
			if(id == null || timeout < 0) {
				logger.trace("no tasks completed within timeout, exiting");
				throw new TaskTimedOutException("imeout expired before all tasks completed");
			}
			logger.debug("task '{}' complete (count: {}, queue: {})", id, count, queue.size());
			T result = futures.get(id).get();
			for(TaskObserver<T> observer : observers) {
				observer.onTaskComplete(result);
			}
			return result;
		}
		return null;
	} 

}